// Copyright 2014 The Go Circuit Project
// Use of this source code is governed by the license for
// The Go Circuit Project, found in the LICENSE file.
//
// Authors:
//   2014 Chris Monson <chris@gocircuit.org>

package ripple

import (
	"bytes"
	"fmt"
	"log"
	"math/rand"
	"reflect"
	"sync"
	"time"

	"circuit/kit/rh"
)

// chanSendFID
type chanSendFID struct {
	rh.AnchoredFID
	sync.Mutex

	hooks  func() chan<- (chan []byte)
	report func(int64)
	q      rh.Q
	gtm    time.Time

	nonblocking bool
	sink        chan []byte
	accumulator *bytes.Buffer
}

// NewChanSend creates a channel send point.
// During Open, an attempt is made to send a data channel to hooks. If
// successful, that data channel can be used to send data to the other end.
// If nb is specified, the operation is non-blocking.
func NewChanSend(hooks func() chan<- (chan []byte), report func(int64), nb bool) rh.FID {
	fid := &chanSendFID{
		sink:        make(chan []byte, 1),
		accumulator: bytes.NewBuffer([]byte{}),
		hooks:       hooks,
		report:      report,
		nonblocking: nb,
		gtm:         time.Now(),
		q: rh.Q{
			ID:  uint64(rand.Int63()),
			Ver: 1,
		},
	}
	return fid
}

func (fid *chanSendFID) String() string {
	return fmt.Sprintf("chansend (%p) %v", fid, fid.Q())
}

func (fid *chanSendFID) Q() rh.Q {
	return fid.q
}

func (fid *chanSendFID) Open(flag rh.Flag, intr rh.Intr) (err error) {
	defer un(lock(fid))

	// avoid blocking simulated opens from the fuse/rh setattr hack
	if flag.Create || flag.Truncate || flag.Deny {
		return rh.ErrPerm
	}

	hooks := fid.hooks()

	if fid.nonblocking {
		hooksValue := reflect.ValueOf(hooks)
		sinkValue := reflect.ValueOf(fid.sink)
		sent := hooksValue.TrySend(sinkValue)
		if !sent {
			return rh.ErrBusy
		}
	} else {
		select {
		case hooks <- fid.sink:
			// Blocks forever if hooks is nil (meaning that capacity is not yet
			// set on the endpoint). That matches normal Go behavior, so we're
			// good.
		case i := <-intr:
			log.Printf("chansend open interrupted: %v", i)
			return rh.ErrGone
		}
	}
	return nil
}

func (fid *chanSendFID) Read(_ int64, count int, intr rh.Intr) (chunk rh.Chunk, err error) {
	return nil, rh.ErrPerm
}

func (fid *chanSendFID) Write(offset int64, data rh.Chunk, intr rh.Intr) (n int, err error) {
	defer un(lock(fid))

	if data == nil {
		// Writing empty data is allowed.
		return 0, nil
	}

	return fid.accumulator.Write(data.(rh.ByteChunk)[offset:])
}

func (fid *chanSendFID) Clunk() {
	defer un(lock(fid))

	select {
	case fid.sink <- fid.accumulator.Bytes():
		close(fid.sink)
		fid.report(int64(fid.accumulator.Len()))
		fid.sink = nil
		fid.accumulator.Reset()
	default:
		log.Printf("Failed to send data when clunking chansend")
	}
}

func (fid *chanSendFID) Walk(wname []string) (rh.FID, error) {
	if len(wname) > 0 {
		return nil, rh.ErrClash
	}
	// Duplicate these guys so that we can have multiple openers.
	return NewChanSend(fid.hooks, fid.report, fid.nonblocking), nil
}

func (fid *chanSendFID) Stat() (*rh.Dir, error) {
	return &rh.Dir{
		Q:      fid.q,
		Mode:   rh.Mode{Attr: rh.ModeIO},
		Perm:   0222, // -w--w--w-
		Atime:  fid.gtm,
		Mtime:  fid.gtm,
		Name:   "chansend",
		Length: 0,
		Uid:    uid(),
		Gid:    gid(),
	}, nil
}
